package com.spark.searchservice.utlis;

import org.frameworkset.elasticsearch.ElasticSearchHelper;
import org.frameworkset.elasticsearch.client.DB2ESImportBuilder;
import org.frameworkset.elasticsearch.client.DataStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author fuqiang
 * @Description 从数据库同步数据到elastic
 * @Date 2021-03-05
 */
public class SynDataFromDbToElastic {

    private static final Logger logger = LoggerFactory.getLogger(SynDataFromDbToElastic.class);

    /**
     * 同步导入es
     */
    public static void simpleImportBuilder(){
        DB2ESImportBuilder importBuilder = DB2ESImportBuilder.newInstance();
        try {
            //清除测试表数据
            ElasticSearchHelper.getRestClientUtil().dropIndice("pms_product");
        }
        catch (Exception e){

        }
        //数据源相关配置，可选项，可以在外部启动数据源
        importBuilder.setDbName("mall")
                // 数据库驱动程序，必须导入相关数据库的驱动jar包
                .setDbDriver("com.mysql.cj.jdbc.Driver")
                //通过useCursorFetch=true启用mysql的游标fetch机制，否则会有严重的性能隐患，useCursorFetch必须和jdbcFetchSize参数配合使用，否则不会生效
                .setDbUrl("jdbc:mysql://localhost:3306/mall?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai&useCursorFetch=true")
                .setDbUser("root")
                .setDbPassword("root")
                .setValidateSQL("select 1")
                .setUsePool(false);//是否使用连接池


        //指定导入数据的sql语句，必填项，可以设置自己的提取逻辑
        importBuilder.setSql("select * from pms_product");
        /**
         * es相关配置
         */
        importBuilder
                .setIndex("pms_product") //必填项
                .setIndexType("pms_product") //必填项
                .setRefreshOption(null)//可选项，null表示不实时刷新，importBuilder.setRefreshOption("refresh");表示实时刷新
                .setUseJavaName(true) //可选项,将数据库字段名称转换为java驼峰规范的名称，例如:doc_id -> docId
                .setBatchSize(5000)  //可选项,批量导入es的记录数，默认为-1，逐条处理，> 0时批量处理
                //设置数据库的查询fetchsize，同时在mysql url上设置useCursorFetch=true启用mysql的游标fetch机制，否则会有严重的性能隐患，
                // jdbcFetchSize必须和useCursorFetch参数配合使用，否则不会生效
                .setJdbcFetchSize(10000);
        /**
         * 执行数据库表数据导入es操作
         */
        DataStream dataStream = importBuilder.builder();
        dataStream.execute();
    }

    /**
     * 异步导入es
     */
    public static void simpleLogImportBuilderFromExternalDBConfig(){
        DB2ESImportBuilder importBuilder = DB2ESImportBuilder.newInstance();
        try {
            //清除测试表
            ElasticSearchHelper.getRestClientUtil().dropIndice("pms_product");
        }
        catch (Exception e){

        }
        //数据源相关配置，可选项，可以在外部启动数据源
        importBuilder.setDbName("mall")
                // 数据库驱动程序，必须导入相关数据库的驱动jar包
                .setDbDriver("com.mysql.cj.jdbc.Driver")
                //通过useCursorFetch=true启用mysql的游标fetch机制，否则会有严重的性能隐患，useCursorFetch必须和jdbcFetchSize参数配合使用，否则不会生效
                .setDbUrl("jdbc:mysql://localhost:3306/mall?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai&useCursorFetch=true")
                .setDbUser("root")
                .setDbPassword("root")
                .setValidateSQL("select 1")
                .setUsePool(false);//是否使用连接池


        //指定导入数据的sql语句，必填项，可以设置自己的提取逻辑
        importBuilder.setSql("select * from pms_product");
        /**
         * es相关配置
         */
        importBuilder
                .setIndex("pms") //必填项
                .setIndexType("pms") //必填项
                .setRefreshOption(null)//可选项，null表示不实时刷新，importBuilder.setRefreshOption("refresh");表示实时刷新
                .setUseJavaName(true) //可选项,将数据库字段名称转换为java驼峰规范的名称，例如:doc_id -> docId
                .setBatchSize(5000)  //可选项,批量导入es的记录数，默认为-1，逐条处理，> 0时批量处理
                //设置数据库的查询fetchsize，同时在mysql url上设置useCursorFetch=true启用mysql的游标fetch机制，
                // 否则会有严重的性能隐患，jdbcFetchSize必须和useCursorFetch参数配合使用，否则不会生效
                .setJdbcFetchSize(10000);

        /**
         * 一次、作业创建一个内置的线程池，实现多线程并行数据导入elasticsearch功能，作业完毕后关闭线程池
         */
        importBuilder.setParallel(true);//设置为多线程并行批量导入
        importBuilder.setQueue(100);//设置批量导入线程池等待队列长度
        importBuilder.setThreadCount(200);//设置批量导入线程池工作线程数量
        importBuilder.setContinueOnError(true);//任务出现异常，是否继续执行作业：true（默认值）继续执行 false 中断作业执行
        //true 异步方式执行，不等待所有导入作业任务结束，方法快速返回；false（默认值） 同步方式执行，等待所有导入作业任务结束，所有作业结束后方法才返回
        importBuilder.setAsyn(false);
        importBuilder.setRefreshOption("refresh"); // 为了实时验证数据导入的效果，强制刷新数据，生产环境请设置为null或者不指定

        /**
         * 执行数据库表数据导入es操作
         */
        DataStream dataStream = importBuilder.builder();
        dataStream.execute();

        long count = ElasticSearchHelper.getRestClientUtil().countAll("pms_product");
        logger.info("数据导入完毕后索引表pms_product中的文档数量: {}",count);
    }
    public static void main(String[] args) {
//        simpleImportBuilder();
        simpleLogImportBuilderFromExternalDBConfig();
    }

}
